package ontime

import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

/**
  * @author dinghao
  * @create 2021-08-04 9:40 
  * @message
  */
object Broadcastflink {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val filterData = env.addSource(new RichSourceFunction[String] {
      override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
        ctx.collect(args(1))
      }
      override def cancel(): Unit = {}
    })


    val configFilter:MapStateDescriptor [String, String] = new MapStateDescriptor[String,String](
      "configFilter", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)

    val broadcastConfig:BroadcastStream[String]= filterData.broadcast(configFilter)

    val data = env.socketTextStream(args(0),9999)
    data.connect(broadcastConfig).process(new BroadcastProcessFunction[String,String,String] {
      private var keyWords = new String
      override def processElement(value: String,
                                  ctx: BroadcastProcessFunction[String, String, String]#ReadOnlyContext,
                                  out: Collector[String]): Unit = {
        if(value.contains(keyWords)){
          out.collect("广播匹配的数据："+value)
        }
      }

      override def processBroadcastElement(value: String, ctx: BroadcastProcessFunction[String, String, String]#Context, out: Collector[String]): Unit = {
        this.keyWords=value
        println("广播的字段为=》" + keyWords)
      }
    }).print()

    env.execute("Broadcastflink")
  }
}
